package com.example.mqtt.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.collection.IntObjectHashMap;
import io.netty.util.concurrent.ScheduledFuture;
import io.netty.util.internal.StringUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import com.example.mqtt.client.MqttPendingObject.*;

/**
 * 无限重连client
 */
@Slf4j
@ChannelHandler.Sharable
public final class RetryMqttClient extends SimpleChannelInboundHandler<MqttMessage> implements IMqttClient {

    private static final EventCallback NOOP = new EventCallback() {};

    private enum CLIENT_STATUS { INIT, CONNECTED, RECONNECT, CLOSED }

    private Channel channel;

    private Bootstrap bootstrap = new Bootstrap();

    /**
     * 事件回调
     */
    private EventCallback eventCallback;

    /**
     * 重试次数
     */
    private AtomicInteger retryCount = new AtomicInteger(0);

    /**
     * 重度策略
     */
    private RetryStrategy retryStrategy;

    private final EventLoopGroup workerGroup = new NioEventLoopGroup();

    private final AtomicInteger nextMessageId = new AtomicInteger(1);
    private AtomicReference<CLIENT_STATUS> status = new AtomicReference<>(CLIENT_STATUS.INIT);

    private Map<Integer, MqttIncomingQos2Publish> incomingQos2Publishes = new IntObjectHashMap<>();
    private Map<Integer, MqttPendingPublish> pendingPublishMap = new IntObjectHashMap<>();
    private final IntObjectHashMap<MqttPendingSubscription> pendingSubscriptions = new IntObjectHashMap<>();
    private final IntObjectHashMap<MqttPendingUnsubscription> pendingServerUnsubscribes = new IntObjectHashMap<>();

    private SocketAddress addr;
    private String username;
    private String password;
    private String clientId;
    private boolean cleanSession;
    private MqttVersion protocolVersion;
    private MqttLastWill lastWill;
    private int keepAliveTimeSeconds;

    private ScheduledFuture<?> pingTimer = null;

    public static class Builder {
        private int connectTimeoutMs = 60000;
        private long readTimeoutMs = 0;
        private Consumer<SocketChannel> onChannelInitializer;
        private EventCallback eventCallback = NOOP;
        private RetryStrategy retryStrategy;

        private String host;
        private int port = 1883;
        private String username;
        private String password;
        private String clientId;
        private boolean cleanSession;
        private MqttVersion protocolVersion = MqttVersion.MQTT_3_1;
        private MqttLastWill lastWill;
        private int keepAliveTimeSeconds = 60;

        public Builder username(String username) {
            this.username = username;
            return this;
        }

        public Builder password(String password) {
            this.password = password;
            return this;
        }

        public Builder clientId(String clientId) {
            this.clientId = clientId;
            return this;
        }

        public Builder cleanSession(boolean cleanSession) {
            this.cleanSession = cleanSession;
            return this;
        }

        public Builder protocolVersion(MqttVersion protocolVersion) {
            this.protocolVersion = protocolVersion;
            return this;
        }

        public Builder lastWill(MqttLastWill lastWill) {
            this.lastWill = lastWill;
            return this;
        }

        public Builder keepAliveTimeSeconds(int keepAliveTimeSeconds) {
            this.keepAliveTimeSeconds = keepAliveTimeSeconds;
            return this;
        }

        public Builder address(String host) {
            this.host = host;
            return this;
        }

        public Builder address(String host, int port) {
            this.host = host;
            this.port = port;
            return this;
        }

        public Builder connectTimeoutMs(int connectTimeoutMs) {
            this.connectTimeoutMs = connectTimeoutMs;
            return this;
        }

        public Builder readTimeoutMs(long readTimeoutMs) {
            this.readTimeoutMs = readTimeoutMs;
            return this;
        }

        public Builder onChannelInitializer(Consumer<SocketChannel> onChannelInitializer) {
            this.onChannelInitializer = onChannelInitializer;
            return this;
        }

        public Builder eventCallback(EventCallback eventCallback) {
            this.eventCallback = eventCallback;
            return this;
        }

        public Builder retryStrategy(RetryStrategy retryStrategy) {
            this.retryStrategy = retryStrategy;
            return this;
        }

        public RetryMqttClient build() {

            if(connectTimeoutMs < 0) {
                throw new IllegalArgumentException("connectTimeoutMs must great than 0");
            }

            if(readTimeoutMs < 0) {
                throw new IllegalArgumentException("readTimeoutMs must great than 0");
            }

            if(eventCallback == null) {
                eventCallback = NOOP;
            }

            if(retryStrategy == null) {
                retryStrategy = RetryStrategy.DEFAULT;
            }

            return new RetryMqttClient(host,
                    port,
                    username,
                    password,
                    clientId,
                    cleanSession,
                    protocolVersion,
                    lastWill,
                    keepAliveTimeSeconds,
                    onChannelInitializer,
                    retryStrategy,
                    connectTimeoutMs,
                    eventCallback);
        }
    }

    private RetryMqttClient(String host,
                            int port,
                            String username,
                            String password,
                            String clientId,
                            boolean cleanSession,
                            MqttVersion protocolVersion,
                            MqttLastWill lastWill,
                            int keepAliveTimeSeconds,
                            Consumer<SocketChannel> onChannelInitializer,
                            RetryStrategy retryStrategy,
                            int connectTimeoutMs,
                            EventCallback eventCallback) {
        this.addr = new InetSocketAddress( host, port );
        this.username = username;
        this.password = password;
        this.clientId = clientId;
        this.cleanSession = cleanSession;
        this.protocolVersion = protocolVersion;
        this.lastWill = lastWill;
        this.keepAliveTimeSeconds = keepAliveTimeSeconds;
        this.eventCallback = eventCallback;
        this.retryStrategy = retryStrategy;
        bootstrap.group(workerGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast("mqttDecoder", new MqttDecoder());
                ch.pipeline().addLast("mqttEncoder", MqttEncoder.INSTANCE);
                if(onChannelInitializer != null) {
                    onChannelInitializer.accept(ch);
                }
                ch.pipeline().addLast(RetryMqttClient.this);
            }
        });
    }

    @Override
    public void connect() {
        doConnect();
    }

    @Override
    public CompletableFuture<Void> publish(String topic, ByteBuf payload, MqttQoS qos, boolean retain) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        if(channel.eventLoop().inEventLoop()) {
            doPublish(future, topic, payload, qos, retain);
        } else {
            channel.eventLoop().execute(() -> doPublish(future, topic, payload, qos, retain));
        }
        return  future;
    }

    @Override
    public CompletableFuture<Void> subscribe(String topic) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        if(channel.eventLoop().inEventLoop()) {
            doSubscribe(future, topic);
        } else {
            channel.eventLoop().execute(() -> doSubscribe(future, topic));
        }
        return future;
    }

    @Override
    public CompletableFuture<Void> unSubscribe(String topic, MqttQoS qos) {
        CompletableFuture<Void> future = new CompletableFuture<>();
        if(channel.eventLoop().inEventLoop()) {
            doUnSubscribe(future, topic, qos);
        } else {
            channel.eventLoop().execute(() -> doUnSubscribe(future, topic, qos));
        }
        return future;
    }

    @Override
    public void close() {
        if(status.getAndUpdate(s -> CLIENT_STATUS.CLOSED) != CLIENT_STATUS.CLOSED) {
            if(channel != null) {
                channel.close();
            }
            workerGroup.shutdownGracefully();
        }
    }

    @Override
    public void close2() {
        channel.close();
    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
        processMsg(ctx, msg);
        eventCallback.onReadMessage(this, ctx, msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        if(status.compareAndSet(CLIENT_STATUS.RECONNECT, CLIENT_STATUS.CONNECTED)) {
            eventCallback.onReconnectSuccess(this);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if(status.get() == CLIENT_STATUS.CLOSED) {
            eventCallback.onClosed(this);
        }
        log.info("channelInactive");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        eventCallback.onExceptionCaught(ctx, cause);
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        super.userEventTriggered(ctx, evt);
        if(evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent) evt;
            switch(event.state()){
                case READER_IDLE:
                    break;
                case WRITER_IDLE:
                    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGREQ, false, MqttQoS.AT_MOST_ONCE, false, 0);
                    channel.writeAndFlush(new MqttMessage(fixedHeader));
                    if(this.pingTimer == null) {
                        this.pingTimer = channel.eventLoop().schedule(() -> {
                            MqttFixedHeader fixedHeader2 = new MqttFixedHeader(MqttMessageType.DISCONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
                            channel.writeAndFlush(new MqttMessage(fixedHeader2)).addListener(ChannelFutureListener.CLOSE);
                        }, this.keepAliveTimeSeconds, TimeUnit.SECONDS);
                    }
                    break;
            }
        }
    }

    private void processMsg(ChannelHandlerContext ctx, MqttMessage msg) {
        switch (msg.fixedHeader().messageType()) {
            case CONNACK:
                handleConack(ctx.channel(), (MqttConnAckMessage) msg);
                break;
            case SUBACK:
                handleSubAck((MqttSubAckMessage) msg);
                break;
            case PUBLISH:
                handlePublish(ctx.channel(), (MqttPublishMessage) msg);
                break;
            case UNSUBACK:
                handleUnSubscribeAck((MqttUnsubAckMessage) msg);
                break;
            case PUBACK:
                handlePubAck((MqttPubAckMessage) msg);
                break;
            case PUBREC:
                handlePubrec(ctx.channel(), msg);
                break;
            case PUBREL:
                handlePubrel(ctx.channel(), msg);
                break;
            case PUBCOMP:
                handlePubcomp(msg);
                break;

            case PINGREQ:
                MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
                channel.writeAndFlush(new MqttMessage(fixedHeader));
                break;
            case PINGRESP:
                if(this.pingTimer != null && !this.pingTimer.isCancelled() && !this.pingTimer.isDone()){
                    this.pingTimer.cancel(true);
                    this.pingTimer = null;
                }
                break;
        }
    }

    private void scheduleConnect(Throwable e) {
        if(status.get() != CLIENT_STATUS.CLOSED && canRetry(e)) {
            if(status.compareAndSet(CLIENT_STATUS.CONNECTED, CLIENT_STATUS.RECONNECT)) {
                eventCallback.onStartReconnect(this);
            }
            workerGroup.schedule(this::doConnect, retryStrategy.backoffStrategy.delayBeforeNextRetry(retryCount.get()), TimeUnit.MILLISECONDS);
        } else {
            close();
        }
    }

    private boolean canRetry(Throwable e) {
        return retryCount.incrementAndGet() < retryStrategy.maxRetry && retryStrategy.retryCondition.shouldRetry(e, retryCount.get());
    }

    private void doConnect() {
        if(status.get() != CLIENT_STATUS.CLOSED) {
            try {
                ChannelFuture f = bootstrap.connect(addr);
                f.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if( !future.isSuccess() ) {
                            eventCallback.onRetryCount(RetryMqttClient.this, retryCount.get());
                            future.channel().close();
                            scheduleConnect(future.cause());
                        } else {
                            retryCount.set(0);
                            channel = future.channel();
                            if(status.compareAndSet(CLIENT_STATUS.INIT, CLIENT_STATUS.CONNECTED)) {
                                eventCallback.onConnected(RetryMqttClient.this);
                            }
                            doLogin();
                            channel.closeFuture().addListener(f -> {
                                scheduleConnect(f.cause());
                            });
                        }
                    }
                });
            } catch (Throwable e) {
                scheduleConnect(e);
            }
        } else {
            log.info("client manual close");
        }
    }

    private void doLogin() {
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader(
                protocolVersion.protocolName(),                       // Protocol Name
                protocolVersion.protocolLevel(),                      // Protocol Level
                !StringUtil.isNullOrEmpty(username),                                    // Has Username
                !StringUtil.isNullOrEmpty(password),                                    // Has Password
                lastWill != null && lastWill.isRetain(),   // Will Retain
                lastWill != null  ? lastWill.getQos().value() : 0,    // Will QOS
                lastWill != null,                           // Has Will
                cleanSession,                                         // Clean Session
                keepAliveTimeSeconds                                  // Timeout
        );
        MqttConnectPayload payload = new MqttConnectPayload(
                clientId,
                lastWill != null ? lastWill.getTopic() : null,
                lastWill != null ? lastWill.getMessage().getBytes() : null,
                username,
                password != null ? password.getBytes() : null
        );
        this.channel.writeAndFlush(new MqttConnectMessage(fixedHeader, variableHeader, payload)).addListener(f -> {
            if(!f.isSuccess()) {
                log.error("send mqtt connect message error", f.cause());
                channel.close();
            }
        });
    }


    private void handleConack(Channel channel, MqttConnAckMessage message) {
        switch (message.variableHeader().connectReturnCode()) {
            case CONNECTION_ACCEPTED:
                if(keepAliveTimeSeconds > 0) {
                    channel.pipeline().addAfter("mqttEncoder", "idleStateHandler", new IdleStateHandler(0, keepAliveTimeSeconds, 0, TimeUnit.SECONDS));
                }
                nextMessageId.set(1);
                incomingQos2Publishes.forEach((k,v) -> v.stopRetryPubRec());
                incomingQos2Publishes.clear();
                pendingPublishMap.forEach((k,v) -> {
                    v.stopRetryPublish();
                    v.stopRetryPubrel();
                });
                pendingPublishMap.clear();
                pendingSubscriptions.forEach((k,v) -> v.stopRetrySubscribe());
                pendingSubscriptions.clear();
                pendingServerUnsubscribes.forEach((k,v) -> v.stopRetryUnSub());
                pendingServerUnsubscribes.clear();
                break;
            case CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD:
            case CONNECTION_REFUSED_IDENTIFIER_REJECTED:
            case CONNECTION_REFUSED_NOT_AUTHORIZED:
            case CONNECTION_REFUSED_SERVER_UNAVAILABLE:
            case CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION:
                this.close();
                break;
        }
        eventCallback.onLoginResult(this, message.variableHeader().connectReturnCode());
    }

    private void doSubscribe(CompletableFuture<Void> future, String topic) {
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.SUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
        MqttTopicSubscription subscription = new MqttTopicSubscription(topic, MqttQoS.AT_MOST_ONCE);
        MqttMessageIdVariableHeader variableHeader = getNewMessageId();
        MqttSubscribePayload payload = new MqttSubscribePayload(Collections.singletonList(subscription));
        MqttSubscribeMessage subMsg = new MqttSubscribeMessage(fixedHeader, variableHeader, payload);
        this.channel.writeAndFlush(subMsg);
        MqttPendingSubscription sub = new MqttPendingSubscription(future, topic,  subMsg);
        pendingSubscriptions.put(variableHeader.messageId(), sub);
        sub.startRetrySubscribe(workerGroup.next(), (msg) -> channel.writeAndFlush(msg));
    }

    private void handleSubAck(MqttSubAckMessage message) {
        MqttPendingSubscription subscription = pendingSubscriptions.remove(message.variableHeader().messageId());
        if(subscription != null) {
            subscription.stopRetrySubscribe();
            subscription.getFuture().complete(null);
        }
    }

    private void doUnSubscribe(CompletableFuture<Void> future, String topic, MqttQoS qos) {
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBSCRIBE, false, MqttQoS.AT_LEAST_ONCE, false, 0);
        MqttMessageIdVariableHeader variableHeader = getNewMessageId();
        MqttUnsubscribePayload payload = new MqttUnsubscribePayload(Collections.singletonList(topic));
        MqttUnsubscribeMessage message = new MqttUnsubscribeMessage(fixedHeader, variableHeader, payload);
        MqttPendingUnsubscription pendingUnsubscription = new MqttPendingUnsubscription(future, topic, message);
        this.pendingServerUnsubscribes.put(variableHeader.messageId(), pendingUnsubscription);
        pendingUnsubscription.startRetryUnSub(workerGroup.next(), (msg) -> channel.writeAndFlush(msg));
    }

    private void handleUnSubscribeAck(MqttUnsubAckMessage message) {
        MqttPendingUnsubscription unsubscription = pendingServerUnsubscribes.remove(message.variableHeader().messageId());
        if (unsubscription != null) {
            unsubscription.stopRetryUnSub();
            unsubscription.getFuture().complete(null);
        }
    }

    private void doPublish(CompletableFuture<Void> future, String topic, ByteBuf payload, MqttQoS qos, boolean retain) {
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retain, 0);
        MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(topic, getNewMessageId().messageId());
        MqttPublishMessage message = new MqttPublishMessage(fixedHeader, variableHeader, payload);
        if(qos == MqttQoS.AT_MOST_ONCE) {
            future.complete(null);
        } else {
            MqttPendingPublish publish = new MqttPendingPublish(variableHeader.packetId(), future, payload.retain(), message, qos);
            publish.startRetryPublish(workerGroup.next(), (msg) -> channel.writeAndFlush(msg));
            MqttPendingPublish old = pendingPublishMap.put(variableHeader.packetId(), publish);
            if(old != null) {
                old.stopRetryPublish();
                old.getFuture().completeExceptionally(new Exception("too many pending publish in queue"));
            }
        }
        this.channel.writeAndFlush(message);
    }

    private void handlePubAck(MqttPubAckMessage message) {
        MqttPendingPublish pendingPublish = pendingPublishMap.remove(message.variableHeader().messageId());
        if(pendingPublish != null) {
            pendingPublish.getFuture().complete(null);
            pendingPublish.stopRetryPublish();
        }
    }

    private void handlePublish(Channel channel, MqttPublishMessage message) {
        switch (message.fixedHeader().qosLevel()) {
            case AT_MOST_ONCE:
                invokeHandlersForIncomingPublish(message);
                break;

            case AT_LEAST_ONCE:
                invokeHandlersForIncomingPublish(message);
                if (message.variableHeader().packetId() != -1) {
                    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
                    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().packetId());
                    channel.writeAndFlush(new MqttPubAckMessage(fixedHeader, variableHeader));
                }
                break;

            case EXACTLY_ONCE:
                if (message.variableHeader().packetId() != -1) {
                    MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 0);
                    MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().packetId());
                    MqttMessage pubrecMessage = new MqttMessage(fixedHeader, variableHeader);

                    MqttIncomingQos2Publish incomingQos2Publish = new MqttIncomingQos2Publish(message.retain(), pubrecMessage);
                    incomingQos2Publishes.put(message.variableHeader().packetId(), incomingQos2Publish);
                    incomingQos2Publish.startRetryPubRec(channel.eventLoop(), (msg) -> this.channel.writeAndFlush(msg));
                    channel.writeAndFlush(pubrecMessage);
                }
                break;
        }
    }

    private void handlePubrec(Channel channel, MqttMessage message) {
        MqttPendingPublish pendingPublish = pendingPublishMap.get(((MqttMessageIdVariableHeader) message.variableHeader()).messageId());
        if(pendingPublish != null) {
            pendingPublish.stopRetryPublish();
            MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0);
            MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) message.variableHeader();
            MqttMessage pubrelMessage = new MqttMessage(fixedHeader, variableHeader);
            channel.writeAndFlush(pubrelMessage);
            pendingPublish.setPubrelMessage(pubrelMessage);
            pendingPublish.startRetryPubrel(workerGroup.next(), channel::writeAndFlush);
        }
    }

    private void handlePubrel(Channel channel, MqttMessage message) {
        MqttIncomingQos2Publish incomingQos2Publish = incomingQos2Publishes.remove(((MqttMessageIdVariableHeader) message.variableHeader()).messageId());
        if (incomingQos2Publish != null) {
            MqttPublishMessage msg = incomingQos2Publish.getIncomingPublish();
            try {
                this.invokeHandlersForIncomingPublish(msg);
            } finally {
                incomingQos2Publish.stopRetryPubRec();
            }
        }
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(((MqttMessageIdVariableHeader) message.variableHeader()).messageId());
        channel.writeAndFlush(new MqttMessage(fixedHeader, variableHeader));
    }

    private void handlePubcomp(MqttMessage message) {
        MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) message.variableHeader();
        MqttPendingPublish pendingPublish = pendingPublishMap.remove(variableHeader.messageId());
        if(pendingPublish != null) {
            pendingPublish.getFuture().complete(null);
            pendingPublish.stopRetryPubrel();
        }
    }

    private void invokeHandlersForIncomingPublish(MqttPublishMessage message) {
        eventCallback.onMessage(message);
    }

    private MqttMessageIdVariableHeader getNewMessageId() {
        this.nextMessageId.compareAndSet(0xffff, 1);
        return MqttMessageIdVariableHeader.from(this.nextMessageId.getAndIncrement());
    }


    @Data
    @SuppressWarnings({"unused", "SimplifiableIfStatement"})
    public static final class MqttLastWill {

        private final String topic;
        private final String message;
        private final boolean retain;
        private final MqttQoS qos;

        public MqttLastWill(String topic, String message, boolean retain, MqttQoS qos) {
            if (topic == null) {
                throw new NullPointerException("topic");
            }
            if (message == null) {
                throw new NullPointerException("message");
            }
            if (qos == null) {
                throw new NullPointerException("qos");
            }
            this.topic = topic;
            this.message = message;
            this.retain = retain;
            this.qos = qos;
        }
    }

    public interface EventCallback {

        /**
         * 第一次连上服务器回调
         * @param client
         */
        default void onConnected(RetryMqttClient client) {
            log.info("onConnected");
        }

        /**
         * 最后断开服务器连接回调
         * @param client
         */
        default void onClosed(RetryMqttClient client) {
            log.info("onClosed");
        }

        /**
         * 开始重连回调
         * @param client
         */
        default void onStartReconnect(RetryMqttClient client) {
            log.info("onStartReconnect");
        }


        default void onRetryCount(RetryMqttClient client, int count) {
            log.info("try: {}", count);
        }

        /**
         * 重连结束回调，
         * @param client
         */
        default void onReconnectSuccess(RetryMqttClient client) {
            log.info("onReconnectSuccess");
        }

        /**
         * 收到消息回调
         * @param client
         * @param ctx
         * @param msg
         */
        default void onReadMessage(RetryMqttClient client, ChannelHandlerContext ctx, Object msg) {}


        default void onExceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            log.error("onExceptionCaught", cause);
        }

        default void onLoginResult(RetryMqttClient client, MqttConnectReturnCode code) {
            log.info("mqtt connect return code: {}", code);
        }

        default void onMessage(MqttPublishMessage msg) {}
    }

    public static class RetryStrategy {

        public static final RetryStrategy DEFAULT = new RetryStrategy(RetryCondition.ALWAYS_RETRY_CONDITION, BackoffStrategy.DELAY_LINEAR, Integer.MAX_VALUE);

        private RetryCondition retryCondition;
        private BackoffStrategy backoffStrategy;
        private int maxRetry;

        public RetryStrategy(RetryCondition retryCondition, BackoffStrategy backoffStrategy, int maxRetry) {
            if(retryCondition == null) {
                retryCondition = RetryCondition.ALWAYS_RETRY_CONDITION;
            }

            if(backoffStrategy == null) {
                backoffStrategy = BackoffStrategy.NO_DELAY;
            }
            if (maxRetry < 0) {
                throw new IllegalArgumentException( "Please provide a non-negative value for maxRetry.");
            }
            this.retryCondition = retryCondition;
            this.backoffStrategy = backoffStrategy;
            this.maxRetry = maxRetry;
        }

        public interface RetryCondition {
            RetryCondition ALWAYS_RETRY_CONDITION = new RetryCondition() {
                @Override
                public boolean shouldRetry(Throwable e, int retryCount) {
                    return true;
                }
            };
            boolean shouldRetry(Throwable e, int retryCount);
        }

        public interface BackoffStrategy {

            public static final BackoffStrategy NO_DELAY = new BackoffStrategy() {
                @Override
                public long delayBeforeNextRetry(int retryCount) {
                    return 0;
                }
            };

            /**
             * 线性增加,每尝试一次增加一秒延迟如[1s,2s,3s.....]
             */
            public static final BackoffStrategy DELAY_LINEAR = new BackoffStrategy() {
                @Override
                public long delayBeforeNextRetry(int retryCount) {
                    return retryCount * 1000L;
                }
            };

            long delayBeforeNextRetry(int retryCount);
        }
    }

}
